package pCloud.obligation;

import java.io.OutputStream;
import java.net.URI;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.Vector;
import java.util.StringTokenizer;

import org.xml.sax.Attributes;
import org.xml.sax.ContentHandler;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;
import org.xml.sax.XMLReader;
import org.xml.sax.helpers.XMLReaderFactory;

import pCloud.PCloudAttributeAndType;
import pCloud.ResourceConnection;
import pCloud.Utilities;
import pCloud.message.DatabaseReply;
import pCloud.message.DatabaseRequest;
import pCloud.message.StreamServerReply;
import pCloud.server.DataStreamHandle;
import pCloud.server.DatabaseHandle;
import pCloud.service.AuroraDataStreamHandle;
import pCloud.stream.*;



import com.streambase.sb.StreamBaseException;
import com.streambase.sb.client.StreamBaseAdminClient;
import com.sun.xacml.Obligation;
import com.sun.xacml.attr.AttributeValue;
import com.sun.xacml.attr.DoubleAttribute;
import com.sun.xacml.attr.IntegerAttribute;
import com.sun.xacml.attr.StringAttribute;
import com.sun.xacml.ctx.Attribute;
import com.sun.xacml.ctx.RequestCtx;

/*
 * Handle Obligations on Data Streams 
 */

/* Phase One: No user query 
 * Step 1. Get request stream name
 * Step 2. Load stream schema from corresponding stream schema file
 * Step 3. Generate FilterBox, MapBox and AggregationBox from Obligation
 * Step 4. Create the SSQL string from stream schema and Boxes
 */

/* Phase Two: with user query 
 * Step 1. Get request stream name
 * Step 2. Load stream schema from corresponding stream schema file
 * Step 3. Generate FilterBoxes, MapBox and AggregationBox from Obligation
 * Step 4. Generate FilterBoxes, MapBox and AggregationBox from User queries
 * Step 5. Try to merge the boxes from step 3 and step 4. Return warnings if there's any
 * Step 6. Create the SSQL string from stream schema and Boxes
 */

class StreamSchemaContentHandler implements ContentHandler{
	
	public Stream stream;
	
	public StreamSchemaContentHandler(QueryGraph qg){
		this.stream = qg.InputStream;		
	}

	public void startElement(String namespaceURI, String localName, String rawName, Attributes atts) {
		
		if (localName.equalsIgnoreCase(PCloudAttributeAndType.STREAM_TAG)){
						
			if (atts.getLength() == 1){
//				System.out.println("Stream Name is " + atts.getValue(0));				
				this.stream.name = atts.getValue(0);
			}
			else {
				System.err.println("XML file invalid, incorrect number of attributes for element " + localName);				
			}
					
		}
		
		if (localName.equalsIgnoreCase(PCloudAttributeAndType.STREAM_ATTRIBUTE_TAG)){
			
			
			if (atts.getLength() == 2) {
				this.stream.attributes.add(new StreamAttribute(atts.getValue(0),atts.getValue(1)));	
			}
			else {
				System.err.println("XML file invalid, incorrect number of attributes for element " + localName);
				
			}
		}		
	}
	
	public void endElement(String namespaceURI, String localName, String rawName) {
		
		if (localName.equalsIgnoreCase("Stream")){
		}
	}
	
	public void startDocument(){
		
		System.out.println("start parsing schema file...");
	}
	
	public void endDocument(){
		
		System.out.println("...Parsing ends...");
	}
	
	public void	endPrefixMapping(String prefix){}

	@Override
	public void characters(char[] ch, int start, int length)
			throws SAXException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void ignorableWhitespace(char[] ch, int start, int length)
			throws SAXException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void processingInstruction(String target, String data)
			throws SAXException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void setDocumentLocator(Locator locator) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void skippedEntity(String name) throws SAXException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void startPrefixMapping(String prefix, String uri)
			throws SAXException {
		// TODO Auto-generated method stub
		
	}
	
}



public class StreamProcessingObligation {
			
	QueryGraph streamquery = new QueryGraph();
	String stream_engine_address;
	String stream_schema_dir;
	int stream_engine_port;
	
	
	public StreamProcessingObligation(){
		
	}
		
	public void loadStreamSchema(String streamName,QueryGraph qg){
		
		String schemaFilePath = this.stream_schema_dir + "/" + streamName + ".xml";
		System.out.println("Schema File for " + streamName + " is \"" + schemaFilePath + "\"");
		
		StreamSchemaContentHandler schemahandler = new StreamSchemaContentHandler(qg);
		try {
			XMLReader parser = XMLReaderFactory.createXMLReader();
			parser.setContentHandler(schemahandler);
			parser.parse(schemaFilePath);				
			
		}
		catch (Exception e){			
			System.out.println(e.getMessage());
		}		
	}
	
	
	public String mergeQueryGraph(){
		
		String mergeResult="";
		
		return mergeResult;
	}
		
	//generate a query graph from user query
	
	public void generateQueryGraphFromQuery(){
		
	}
	
	/*
	 * Construct FilterBox from policy obligation
	 */
	
	public FilterBox constructFilterBox(Obligation ob) throws Exception{
		
		FilterBox fb = new FilterBox();
		List attrs = ob.getAssignments();
		Attribute attr;
		
		if (attrs.size()!=1){
			throw new Exception("Incorrect Number of Attributes for " + ob.getId());
		}
		attr = (Attribute)attrs.get(0);
		if (attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_FILTERING))){
			fb.filtercondition = ((StringAttribute)attr.getValue()).getValue();
		}
		else {
			throw new Exception("Unrecognized attribute " + attr.getId());
		}
	//	fb.print();
		return fb;
	}
	
	/*
	 * Construct MapBox from policy obligation
	 */
	
	public MapBox constructMapBox(Obligation ob)throws Exception{
		
		MapBox mb = new MapBox();
		
		List attrs = ob.getAssignments();
		Attribute attr;
		
		if (attrs.isEmpty()){
			throw new Exception("Incorrect Number of Attributes for " + ob.getId());
		}
		for (int i=0;i<attrs.size();i++){
			
			attr = (Attribute)attrs.get(i);
			if (attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_MAPPING))){
				mb.attributes.add(((StringAttribute)attr.getValue()).getValue());
			}
			else {
				throw new Exception("Unrecognized attribute " + attr.getId());
			}
		}
		
		return mb;
	}
	
	/*
	 * Construct AggregationBox from policy obligation
	 */
	
	public AggregateBox constructAggregateBox(Obligation ob)throws Exception{
		
		AggregateBox ab = new AggregateBox();
		
		List attrs = ob.getAssignments();
		Attribute attr;
		long iattr;
		if (attrs.isEmpty()){
			throw new Exception("Incorrect Number of Attributes for " + ob.getId());
		}
		
		for(int i=0; i<attrs.size();i++){
			
			attr = (Attribute)attrs.get(i);			
			if (attr.getValue() instanceof IntegerAttribute) {
				iattr = ((IntegerAttribute) attr.getValue()).getValue();
				if (attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_AGGEGATION_WINDOW_SIZE))){
					ab.windowsize = (int) iattr;
					
				}
				else if (attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_AGGEGATION_WINDOW_STEP))){
					ab.windowadvancestep = (int) iattr;
					
				}
				else {
					throw new Exception("Unrecognized attribute " + attr.getId());
				}
			}
			else if (attr.getValue() instanceof StringAttribute) {
				String value = ((StringAttribute) attr.getValue()).getValue();
				if (attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_AGGEGATION_WINDOW_TYPE))) {
					ab.windowtype = value;
				}
				else if(attr.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_ATT_AGGEGATION_WINDOW_ATTR))){
					StringTokenizer st = new StringTokenizer(value,":");
					String a_attribute = st.nextToken();
					String a_function = st.nextToken();
					ab.aggregatefunction.add(a_function);
					ab.aggregateattribute.add(a_attribute);					
				}
				else {
					throw new Exception("Unrecognized attribute " + attr.getId());
				}
			}
		}
	//	ab.print();
		return ab;
	}
	
	
	/*
	 * Generate the SSQL scripts to be sent to the stream engine. 
	 * It requires the obligation from the policy, and the stream name got from requestCtx.
	 * Once the stream name is retrieved, corresponding stream schema is obtained from schema file.
	 * 
	 */
	
	public String generateSSQL(Set<Obligation> obligations,RequestCtx requestCtx) throws Exception{
						
		
		ArrayList<AttributeValue> requestedStreams = Utilities
				.getAttributeValue(PCloudAttributeAndType.RESOURCE_TABLE_ID,
						requestCtx.getResource());
		
		String streamname = ((StringAttribute) requestedStreams.get(0)).getValue();
		
//		System.out.println("stream name is " + streamname);
		
		
		Vector<FilterBox> fbs = new Vector<FilterBox>();
		Vector<MapBox> mbs = new Vector<MapBox>();
		Vector<AggregateBox> abs = new Vector<AggregateBox>(); 
		
		Iterator it = obligations.iterator();
		while (it.hasNext()){		
			Obligation ob = (Obligation)it.next();
			if(ob.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_FILTERING_ID))){
				fbs.add(this.constructFilterBox(ob));
			}
			else if(ob.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_MAPPING_ID))){
				mbs.add(this.constructMapBox(ob));
			}
			else if(ob.getId().equals(new URI(PCloudAttributeAndType.OBLIGATION_STREAM_WINDOW_AGGREGATION_ID))){
				abs.add(this.constructAggregateBox(ob));
			}
			else {
				throw new Exception("Unrecognized obligation " + ob.getId());
			}
				
		}
		
	//	System.out.println("Boxes constructed, now loading stream schema...");
						
		this.loadStreamSchema(streamname, this.streamquery);
		
		System.out.println("Stream schema loaded, now form query graph...");
		this.generateQueryGraphFromObligation(fbs,mbs,abs,streamquery);
		System.out.println("Query graph generated, now generating ssql...");
				
		return streamquery.createSSQL();
				
	}
	
	public void generateQueryGraphFromObligation(Vector<FilterBox> filters, Vector<MapBox> maps, Vector<AggregateBox> aggregates, QueryGraph streamquery){
		
		streamquery.AB = aggregates;
		streamquery.FB = filters;
		streamquery.MB = maps;
				
	}	
	
	public String addQueryGraph(String ssql, String outputStream) throws StreamBaseException{
		
		String uri = "sb://" + this.stream_engine_address + ":" + this.stream_engine_port;
		AuroraDataStreamHandle.container_index++;
		String msg = "";
				
		
		StreamBaseAdminClient sba = new StreamBaseAdminClient(uri);
		String container = "PCLOUD_CONTAINER_" + AuroraDataStreamHandle.container_index;
			
		System.out.println("Adding SSQL In Container : " + container + "...");
			
		sba.addContainer(container, ssql.getBytes(), new Vector<String>());
			
			// msg is URI for streambase client to connect to get output stream
			
		msg = uri+"/"+container+"/"+outputStream;
		
		System.out.println("msg is " + msg);
		return msg;
	}
	
	public Object processObligation(DataStreamHandle ads,Set<Obligation> obligations,RequestCtx requestCtx){
		
		StreamServerReply ssr = new StreamServerReply();
		String ssql = "";
		try {
			long starttime = System.currentTimeMillis();
			ssql = this.generateSSQL(obligations, requestCtx);
			long endtime = System.currentTimeMillis();
			
			double qgtime=(endtime - starttime)*1.0/1000;
			System.out.println("qg time :" + qgtime);
	//		System.out.println(ssql);
			ssr.qgtime=qgtime;
		}
		catch (Exception e){
			System.out.println(e.getMessage());			
		}
		
		try {
			long starttime = System.currentTimeMillis();
			ssr.URI=this.addQueryGraph(ssql, this.streamquery.OutputStreamName);
			long endtime = System.currentTimeMillis();
			double sbtime=(endtime - starttime)*1.0/1000;
			System.out.println("sb time :" + sbtime);
			ssr.sbtime = sbtime;
		}
		catch (StreamBaseException sbe) {
			
			System.out.println(sbe.getMessage());
			ssr.ErrorFromSB = sbe.getMessage();
		}
		
		return ssr;
		
	}
}
